package com.dahuan.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;

public class MyConsumer {

    public static void main(String[] args) {

        //TODO 1 创建kafka 配置信息
        Properties properties = new Properties();

        //TODO 2 给配置信息赋值
        properties.put( ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux-1:9092" ); //连接的集群
        properties.put( ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true ); //是否启动自动提交配置
        properties.put( ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000" ); //自动提交延迟 （毫秒为单位），提交到偏移量 （offsets）

        //TODO 3 KV 的反序列化
        properties.put( ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer" );
        properties.put( ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer" );

        //TODO 4 消费者组
        properties.put( ConsumerConfig.GROUP_ID_CONFIG, "huan" );

        //TODO 5 重置消费者的offset
        properties.put( ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest" );


        //TODO 创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>( properties );

        //TODO 只能用一个集合
        // consumer.subscribe( Collections.singletonList( "first" ) );
        //TODO 多个topic
        // consumer.subscribe( Arrays.asList( "first","second","three" ) );

        //TODO 订阅主题 （topic）
        consumer.subscribe( Collections.singletonList( "first" ) );

        while (true) {
            //TODO 获取数据
            ConsumerRecords<String, String> consumerRecords = consumer.poll( 100 );

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {

                System.out.println( consumerRecord.key() + "--" + consumerRecord.value() );

            }
        }
    }
}
